package com.zenitera.bigdata.transform;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * 对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream
 */
public class Flink11_TransForm_Union {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> stream_01 = env.fromElements(1, 2, 3);
        DataStreamSource<Integer> stream_02 = env.fromElements(4, 5, 6);
        DataStreamSource<Integer> stream_03 = env.fromElements(7, 8, 9);
        DataStreamSource<String> stream_04 = env.fromElements("aaa", "bbb", "ccc");
        DataStreamSource<String> stream_05 = env.fromElements("ddd", "eee", "fff");
        DataStreamSource<String> stream_06 = env.fromElements("ggg", "hhh", "iii");

        stream_01
                .union(stream_02)
                .union(stream_03)
                .print();

        stream_04
                .union(stream_05)
                .union(stream_06)
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
